package rabbitmq

import (
	"encoding/json"
	"github.com/spf13/viper"
	"github.com/streadway/amqp"
	"tg-bot/common"
)

var MQConn *amqp.Connection
var RabbitChannel *amqp.Channel
var consumer string
var ConsumerMap *RabbitContext

func InitRabbitMQConfig() {
	if !viper.IsSet(mq_key) {
		common.SysError("System Not Setting Rabbitmq Config")
		return
	}
	url := viper.GetString(mq_key)
	var err error
	MQConn, err = amqp.Dial(url)
	if err != nil {
		common.SysError("System Config RabbitMQ Fail")
		return
	}
	RabbitChannel, err = MQConn.Channel()
	if err != nil {
		common.SysError("Get Rabbitmq Channal Fail")
		return
	}
	bindRabbitmq()
	common.SysLog("Success Init RabbitMQ Config")
}

func CloseMQConnection() {
	if RabbitChannel != nil {
		err := RabbitChannel.Close() // 先关闭通道
		if err != nil {
			common.SysError("Failed to close RabbitMQ channel: " + err.Error())
		}
	}
	if MQConn != nil {
		err := MQConn.Close() // 然后关闭连接
		if err != nil {
			common.SysError("Failed to close RabbitMQ connection:" + err.Error())
		}
	}
}

// 构建队列
func bindRabbitmq() {
	for _, data := range exchangeVoList {
		var err error
		if data.Type == 0 {
			err = buildBasicExchange(data)
		} else {
			err = buildDelayExchange(data)
		}
		if err != nil {
			continue
		}
		buildBasicQueue(data.QueueList, data.Name, data.Type)
	}
}

func buildBasicQueue(QueueList []QueueRouterVo, exchange string, isDelay int) {
	for _, data := range QueueList {
		args := amqp.Table{}
		if isDelay == 1 {
			args = amqp.Table{
				"x-dead-letter-exchange":    exchange,
				"x-dead-letter-routing-key": data.RouterKey,
			}
		}
		queue, err := RabbitChannel.QueueDeclare(
			data.Name,
			true,
			false,
			false,
			false,
			args,
		)
		if err != nil {
			common.SysError("Queue Declare Failed: " + err.Error())
			continue
		}
		err = RabbitChannel.QueueBind(queue.Name, data.RouterKey, exchange, false, nil)
		if err != nil {
			common.SysError("Queue Bind error: " + err.Error())
			continue
		}
		// 获取所有策略指针
		ConsumerMap = getRabbitMQAllContext()
		buildConsumer(data)
	}
}

func buildBasicExchange(data ExchangeVo) error {
	err := RabbitChannel.ExchangeDeclare(
		data.Name, EXCHANGE_TYPE, true, false, false, false, nil,
	)
	return err
}
func buildDelayExchange(data ExchangeVo) error {
	err := RabbitChannel.ExchangeDeclare(
		data.Name, "x-delayed-message", true, false, false, false,
		amqp.Table{
			"x-delayed-type": EXCHANGE_TYPE,
		},
	)
	return err
}

// 发送消息
func SendMQMessage(data QueueBo) {
	if !viper.IsSet(mq_key) {
		return
	}
	marshal, _ := json.Marshal(data)
	pushType := amqp.Publishing{
		ContentType: "application/json",
		Body:        marshal,
	}
	if data.IsDelayed {
		pushType.Headers = amqp.Table{"x-delay": int64(data.DelayTime)}
	}
	err := RabbitChannel.Publish(
		data.Exchange, data.RoutingKey, false, false, pushType,
	)
	if err != nil {
		common.SysError("RabbitMQ SendMessage Failed, RoutingKey: " + data.RoutingKey + "Error :" + err.Error())
	}
}

// 构建消费者
func buildConsumer(queue QueueRouterVo) {
	queueMsg, err :=
		RabbitChannel.Consume(queue.Name, "", false, false, false, false, nil)
	if err != nil {
		common.SysError("Receive Failed: " + err.Error())
		return
	}
	go func() {
		for msg := range queueMsg {
			var bo QueueBo
			err := json.Unmarshal(msg.Body, &bo)
			if err != nil {
				common.SysError("Unmarshal Failed: " + err.Error())
				msg.Ack(false)
				continue
			}
			ConsumerMap.getAppointConsumer(queue.Name, bo)
			msg.Ack(false)
		}
	}()
}

func getRabbitMQAllContext() *RabbitContext {
	return &RabbitContext{
		consumerMap: map[string]ConsumeStrategy{
			QUEUE_LOGGER:  &LoggerConsumer{},
			QUEUE_DLETE:   &DeleteConsumer{},
			QUEUE_DELAYED: &DelayConsumer{},
		},
	}
}
func (ctx *RabbitContext) getAppointConsumer(contextName string, msg QueueBo) {
	if strategy, ok := ctx.consumerMap[contextName]; ok {
		strategy.Consume(msg)
	}
}
